package org.lql.chaos.kafka.example.consumer;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

/**
 * Kafka 消费者模板（大而全）
 * 包含常见的 @KafkaListener 参数及监听方法参数配置，用于参考或复用
 */
@Service
public class FullKafkaConsumerTemplate {

	private static final Logger log = LoggerFactory.getLogger(FullKafkaConsumerTemplate.class);

	/**
	 * Kafka 消费方法
	 */
	@KafkaListener(
			id = "fullListenerId",                        // 监听器 ID，可用于容器控制
			topics = {"topic-A", "topic-B"},              // 监听的多个 topic
			groupId = "full-listener-group",              // 消费组 ID，可覆盖 application.yml 中的配置
			topicPattern = "",                            // 可选：正则匹配 topic（与 topics 二选一）
			concurrency = "3",                            // 开启 3 个线程并发处理（注意 topic 分区数限制）
			autoStartup = "true",                         // 自动启动监听器（false 表示手动控制）
			clientIdPrefix = "my-client-",                // Kafka 客户端 ID 前缀，会自动加上编号
			containerFactory = "customKafkaContainerFactory", // 使用自定义容器工厂（可支持 JSON、批量等）
			properties = {
					"max.poll.records:100",                   // 覆盖某些消费者属性
					"fetch.min.bytes:1024"
			}
	)
	public void onMessage(
			ConsumerRecord<String, String> record,                      // Kafka 原始记录（包含所有元数据）
			@Payload String payload,                                    // 消息体内容（已反序列化）
			@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,      // 消息键
			@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,          // 消息 topic
			@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,  // 分区号
			@Header(KafkaHeaders.OFFSET) long offset,                   // 偏移量
			@Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp,    // 消息时间戳
			Headers headers,                                            // Kafka 原生 Headers，可用于获取自定义头
			Acknowledgment ack,                                         // 手动 ack 控制 offset 提交
			Consumer<?, ?> consumer                                     // Kafka 原生 Consumer，可用于 commitAsync 等操作
	) {
		// ========= 打印基本信息 =========
		log.info(">>> Received message from topic={}, partition={}, offset={}, timestamp={}",
				topic, partition, offset, timestamp);
		log.info(">>> Key={}, Payload={}", key, payload);

		// ========= 处理自定义 header =========
		org.apache.kafka.common.header.Header customHeader = headers.lastHeader("some-custom-header");
		if (customHeader != null) {
			log.info(">>> Custom Header [some-custom-header] = {}", new String(customHeader.value()));
		}

		// ========= 打印所有 header =========
		for (org.apache.kafka.common.header.Header header : headers) {
			log.info(">>> Header: {} = {}", header.key(), new String(header.value()));
		}

		// ========= 业务逻辑处理 =========
		try {
			// TODO: 你的业务逻辑
			log.info(">>> Processing business logic...");

			// ========= 手动 ack =========
			ack.acknowledge();

		} catch (Exception e) {
			log.error("!!! Error while processing message from topic={}, offset={}", topic, offset, e);
			// ❗ 未 ack 会导致消息重试（取决于 Kafka 配置）
		}
	}
}